Skip to main content

安装

安装包下载​

链接：https://pan.baidu.com/s/1tURBIZCZEh6Bp_H8r_iGsQ 
提取码：yyds 
--来自百度网盘超级会员V5的分享

安装​

解压​

tar -zxvf kafka_2.12-3.6.0.tgz -C ../module/

创建数据存放目录

mkdir -p /datadrive/kafka/log

修改配置文件​

server.properties​

vi /home/bigdata/module/kafka_2.12-3.6.0/config/server.properties
# 每台机器的id不一样
broker.id=0

# 核心数加1
num.network.threads=9

# 核心数乘2
num.io.threads=16

# 默认100M，修改成2倍，增加网络缓冲，提高吞吐量
socket.send.buffer.bytes=204800

# 默认100M，修改成2倍，增加网络缓冲，提高吞吐量
socket.receive.buffer.bytes=204800

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600

# 副本从leader抓取消息的线程数目，默认是1
num.replica.fetchers=3

# 最小同步副本数
min.insync.replicas=2

# 数据到期删除策略
log.cleanup.policy=delete

#默认值是1M-1048576 (1 mebibyte)，设置成5M
message.max.bytes=5242880

replica.fetch.min.bytes=1

#分区副本同步最大的消息默认1M-1048576 (1 mebibyte)，设置成5M
replica.fetch.max.bytes=5242880

# 自动创建主题
auto.create.topics.enable=true

# 默认分区副本
default.replication.factor=3


############################# Log Basics #############################

# A comma separated list of directories under which to store log files
log.dirs=/datadrive/kafka/log

# 默认分区
num.partitions=5

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=3

############################# Internal Topic Settings  #############################
#表示Kafka的内部topic __consumer_offsets只有一个副本。当该副本所在的broker宕机时，__consumer_offsets只有一份副本，该分区就会宕机。使用该分区存储消费分组offset位置的消费者均会收到影响，offset无法提交，从而导致生产者可以发送消息但消费者不可用。
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3

############################# Log Flush Policy #############################


# 每当producer写入10000条消息时，刷数据到磁盘
log.flush.interval.messages=10000
# 每间隔1秒钟时间，刷数据到磁盘
log.flush.interval.ms=1000

############################# Log Retention Policy #############################

# 日志保留时长 7 天
log.retention.hours=168

# 段文件配置
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

############################# Zookeeper #############################

zookeeper.connect=master1:2181,master2:2181,node1:2181/kafka
zookeeper.connection.timeout.ms=18000

############################# Group Coordinator Settings #############################

# 消费者组再均衡的延迟，默认是3秒
group.initial.rebalance.delay.ms=3

kafka-server-start.sh​

修改配置文件

vi /home/bigdata/module/kafka_2.12-3.6.0/bin/kafka-server-start.sh
KAFKA_HEAP_OPTS="-Xmx16G -Xms16G"

启动​

重启前修改其他集群机器​

分发kafka到所有的机器

./xsync.sh /home/bigdata/module/kafka_2.12-3.6.0

分别修改机器的

vi /home/bigdata/module/kafka_2.12-3.6.0/config/server.properties

# 每台机器的id不一样
broker.id=1

集群启停脚本​

vi kafka_cluster.sh
#!/bin/bash
 
case $1 in
"start"){
    for i in master1 master2 node1
    do
        echo " --------启动 $i Kafka-------"
        ssh $i "/home/bigdata/module/kafka_2.12-3.6.0/bin/kafka-server-start.sh -daemon /home/bigdata/module/kafka_2.12-3.6.0/config/server.properties "
    done
};;
"stop"){
    for i in master1 master2 node1
    do
        echo " --------停止 $i Kafka-------"
        ssh $i "/home/bigdata/module/kafka_2.12-3.6.0/bin/kafka-server-stop.sh stop"
    done
};;
esac
chmod 744 kafka_cluster.sh 
./kafka_cluster.sh start

网络相关配置​

不配置listeners的情况​

#listeners = PLAINTEXT://your.hostname:9092

在zookeeper中，/brokers/{brokerid}/下面的信息示例

{
    "listener_security_protocol_map": {
        "PLAINTEXT": "PLAINTEXT"
    },
    "endpoints": ["PLAINTEXT://master1:9092"],
    "jmx_port": -1,
    "features": {},
    "host": "master1",
    "timestamp": "1698675472994",
    "port": 9092,
    "version": 5
}

那么如果其他的机器想访问，那么就需要配置/etc/hosts指定对应的域名，对应上面的master1才能够访问，比如

vi /etc/hosts

master1 xxx.xxx.xx(ip)

内外网访问的情况​

下面 19002，19001，19000是同一个外网ip不同端口对集群的访问配置。

机器1
listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:19002
advertised.listeners=INTERNAL://内网ip:9092,EXTERNAL://外网ip:19002
listener.security.protocol.map=EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
inter.broker.listener.name=INTERNAL

机器2
listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:19001
advertised.listeners=INTERNAL://内网ip:9092,EXTERNAL://外网ip:19001
listener.security.protocol.map=EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
inter.broker.listener.name=INTERNAL

机器3
listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:19000
advertised.listeners=INTERNAL://内网ip:9092,EXTERNAL://外网ip:19000
listener.security.protocol.map=EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
inter.broker.listener.name=INTERNAL

假设您在 listeners 中配置了不同的端口，例如 EXTERNAL://0.0.0.0:5001，而在 advertised.listeners 中配置了 EXTERNAL://外网ip:5000，那么 Kafka 代理将监听 5001 端口，但客户端连接时将使用 5000 端口，这将导致连接失败。因此，在同一监听器下，listeners 和 advertised.listeners 中的端口号必须相同，以确保客户端能够成功连接 Kafka 代理。

相关链接

https://bbs.huaweicloud.com/blogs/346008

Kafka每一台机器都有外网地址的时候​

listeners=INTERNAL://0.0.0.0:9092
advertised.listeners=INTERNAL://172.16.7.237:9092
listener.security.protocol.map=EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
inter.broker.listener.name=INTERNAL